package cn.me.alphamq.common.model.broker.meta;

import cn.me.alphamq.common.lock.WriteDataLock;
import cn.me.alphamq.common.lock.UnfairReentryLock;
import cn.me.alphamq.common.util.FileNameUtil;
import cn.me.alphamq.common.util.FileUtil;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * MMap 映射磁盘的文件的模型，仅可以对 CommitLog 和 ConsumeQueue 的文件使用
 * <p>
 * 本对象有对应的文件的指针 fileModel，当创建新文件时，使用它来修改文件的信息
 *
 * @author f
 */
public class MMapFileModel {

    /**
     * 映射文件的目录路径
     */
    private String fileDir;

    /**
     * 映射到内存中的数据的最大体积
     */
    private int maxSize;

    /**
     * 映射到磁盘的 buffer，用于写入数据
     */
    private MappedByteBuffer mappedBuffer;

    /**
     * 用于读取数据的 buffer，与写入数据的 buffer 的 position 不同，避免写入数据影响读取数据，指向的是同一块 buffer
     * <p>
     * 这个 buffer 的 position 是不变的，读取数据时只需要再分出一个切片即可
     */
    private ByteBuffer readBuffer;

    /**
     * 写入数据时加的锁，防止多线程同时写入覆盖数据
     */
    private WriteDataLock writeDataLock;

    /**
     * 本 MMap 映射文件在内存中对应的对象
     */
    private MappedFile fileModel;

    /**
     * 将指定文件指定起始位置之后的指定长度的数据映射到内存中，初始化本对象
     *
     * @param file        文件
     * @param fileDir     文件目录路径
     * @param maxSize     映射文件的最大大小
     * @throws IOException
     */
    public void mapFile(MappedFile file, String fileDir, int maxSize) throws IOException {
        this.fileDir = fileDir;
        this.fileModel = file;
        this.maxSize = maxSize;
        // 默认使用非公平锁
        this.writeDataLock = new UnfairReentryLock();
        doMMap();
    }

    /**
     * 从本文件指定 offset 的位置开始，读取指定长度的内容
     *
     * @param startOffset 起始的偏移量
     * @param length      读取数据的长度
     * @return
     */
    public byte[] read(int startOffset, int length) {
        // 分出一个切片供读取数据使用
        ByteBuffer readBuffer = this.readBuffer.slice();
        readBuffer.position(startOffset);
        byte[] content = new byte[length];
        readBuffer.get(content);
        return content;
    }

    /**
     * 以指定的方式将数据写入文件，可以选择同步刷盘还是异步刷盘
     *
     * @param data  数据
     * @param force 是否强制同步刷盘
     * @return 返回本数据在文件中的偏移量
     * @throws IOException
     */
    public int write(byte[] data, boolean force) throws Exception {
        try {
            writeDataLock.lock();

            ensureEnoughSize(data.length);

            mappedBuffer.put(data);

            int offset = fileModel.getOffset().getAndAdd(data.length);

            // 默认刷到 page cache 中（异步刷盘）
            if (force) {
                // 强制刷盘
                mappedBuffer.force();
            }

            return offset;
        } finally {
            writeDataLock.unlock();
        }
    }

    /**
     * 确保文件有足够的大小写入新数据
     *
     * @param need 需要的大小
     */
    private void ensureEnoughSize(int need) throws Exception {
        // 如果文件有足够的大小，则直接返回
        if (fileModel.restSize() >= need) {
            return;
        }

        // 创建一个新的文件
        // 修改文件模型的信息
        fileModel.setFileName(FileNameUtil.getNewFileName(fileModel.getFileName()));
        fileModel.setOffsetLimit(maxSize);
        fileModel.setOffset(new AtomicInteger(0));

        String newFilePath = fileDir + fileModel.getFileName();
        File newFile = new File(newFilePath);
        try {
            newFile.createNewFile();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        // 映射这个新文件
        doMMap();
    }

    /**
     * 释放 MMap 内存占用
     */
    public void clean() {
        if (mappedBuffer == null || !mappedBuffer.isDirect() || mappedBuffer.capacity() == 0) {
            return;
        }
        // 安全地（有权限的？）调用 DirectByteBuffer 对象中的 cleaner 字段的 clean() 方法
        invoke(invoke(viewed(mappedBuffer), "cleaner"), "clean");
    }

    // 通过 DirectByteBuffer 对象中的 att 字段，来获取最终的 DirectByteBuffer
    private ByteBuffer viewed(ByteBuffer buffer) {
        String methodName = "viewedBuffer";
        Method[] methods = buffer.getClass().getMethods();
        for (Method method : methods) {
            if ("attachment".equals(method.getName())) {
                methodName = "attachment";
                break;
            }
        }

        // 如果 ByteBuffer 的 att 字段没有值，则说明这个 ByteBuffer 不是 DirectByteBuffer
        // 它的 att 字段指向的 ByteBuffer 可能是 DirectByteBuffer
        ByteBuffer viewedBuffer = (ByteBuffer) invoke(buffer, methodName);
        if (viewedBuffer == null) {
            return buffer;
        } else {
            return viewed(viewedBuffer);
        }
    }

    // 在安全的环境下调用指定方法
    private Object invoke(final Object target, final String methodName, final Class<?>... args) {
        return AccessController.doPrivileged(new PrivilegedAction<Object>() {
            @Override
            public Object run() {
                try {
                    Method method = method(target, methodName, args);
                    method.setAccessible(true);
                    return method.invoke(target);
                } catch (Exception e) {
                    throw new IllegalStateException(e);
                }
            }
        });
    }

    // 获取 指定对象拥有的指定名称和参数 的方法
    private Method method(Object target, String methodName, Class<?>... args) throws NoSuchMethodException {
        try {
            return target.getClass().getMethod(methodName, args);
        } catch (NoSuchMethodException e) {
            return target.getClass().getDeclaredMethod(methodName, args);
        }
    }

    /**
     * 将本文件指定起始位置之后的指定长度的数据映射到内存中
     *
     * @throws IOException
     */
    private void doMMap() throws IOException {
        String filePath = fileDir + fileModel.getFileName();
        File file = new File(filePath);
        if (!file.exists()) {
            // 如果一开始没有对应的目录或文件，则创建文件及目录
            FileUtil.createFile(fileDir, FileNameUtil.generateFileName());
        }
        mappedBuffer = new RandomAccessFile(file, "rw").getChannel()
                .map(FileChannel.MapMode.READ_WRITE, 0, maxSize);
        // 将 mappedBuffer 的切片作为 读取数据时使用的 buffer
        readBuffer = mappedBuffer.slice();
        mappedBuffer.position(fileModel.getOffset().get());
    }

    public MappedFile getFileModel() {
        return fileModel;
    }
}
